package net.xinqushi.design.templateMethod.core;

import net.xinqushi.design.templateMethod.message.Message;
import net.xinqushi.design.templateMethod.pool.ConnectionFactory;
import net.xinqushi.design.templateMethod.pool.ConnectionPool;
import net.xinqushi.design.templateMethod.pool.GenericConnectionPool;
import net.xinqushi.design.templateMethod.pool.SocketConnectionFactory;
import net.xinqushi.design.templateMethod.util.NumberUtils;

import java.io.ByteArrayOutputStream;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;



public  class PoolableSocketMessageSender extends SocketMessageSender implements ConnectionFactory {

	@SuppressWarnings("rawtypes")
	protected volatile ConnectionPool pool=null;

	protected int maxActive = GenericConnectionPool.DEFAULT_MAX_ACTIVE;

	protected long maxWait = GenericConnectionPool.DEFAULT_MAX_BLOCKING_TIME;

	protected long keepIdleTime = GenericConnectionPool.DEFAULT_MAX_IDLE_TIME;

	protected AtomicBoolean usable=new AtomicBoolean(false);

	protected AtomicInteger workCounter=new AtomicInteger();

	public PoolableSocketMessageSender(String host,int port,String username, String password) {
		super(host, port,username,password);
	}

	public int getMaxActive() {
		return maxActive;
	}

	public void setMaxActive(int maxActive) {
		this.maxActive = maxActive;
	}

	public long getMaxWait() {
		return maxWait;
	}

	public void setMaxWait(long maxWait) {
		this.maxWait = maxWait;
	}

	public long getKeepIdleTime() {
		return keepIdleTime;
	}

	public void setKeepIdleTime(long keepIdleTime) {
		this.keepIdleTime = keepIdleTime;
	}

	protected synchronized void initialize() throws Exception{
		if(initialized)
			return;
		if(pool==null){
			pool=createConnectionPool();
		}
		initialized=true;
		usable.set(true);
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public ConnectionPool createConnectionPool() throws Exception {
		SocketConnectionFactory factory=createPoolableConnectionFactory();
		ConnectionPool pool=new GenericConnectionPool(factory,maxActive,maxWait,keepIdleTime);
		return pool;
	}

	public SocketConnectionFactory createPoolableConnectionFactory()
	throws Exception {
		return new SocketConnectionFactory(host,port,username,password);
	}

	@Override
	protected void onPreSent() throws Exception {
		if(!initialized){
			initialize();
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	protected String messageSent(String message) throws Exception {
		if(!usable.get()){
			return "401";
		}
		workCounter.incrementAndGet();
		String returnVal="402";
		try{
			Socket socket=(Socket) pool.getConnection();
			socket.setSoTimeout(readTimeout);
			ByteArrayOutputStream write_buff=new ByteArrayOutputStream();//实现socket不关闭一直生效，原因待探究
			byte[] bytes=message.getBytes();
			write_buff.write(NumberUtils.convertIntegerToByteArray(bytes.length));
			write_buff.write(bytes);
			write_buff.writeTo(socket.getOutputStream());
			write_buff.close();
			byte[] read_buff=new byte[readBufferSize];
			int readBytes=0;
			while(true){
				if((readBytes=socket.getInputStream().read(read_buff))>0){
					break;
				}
			}
			write_buff=new ByteArrayOutputStream();
			write_buff.write(read_buff, 0, readBytes);
			returnVal=write_buff.toString();
			write_buff.close();
			pool.returnConnection(socket);
		}catch(Exception e){
			
		}
		return returnVal;
	}

	@Override
	protected void onPostSent() throws Exception {
		workCounter.decrementAndGet();
	}

	public void completeSending(){
		usable.getAndSet(false);
		try {
			while(true){
				if(workCounter.get()==0){
					break;
				}
			}
			pool.close();
		} catch (Exception e) {

		}
	}

}
